package com.doeis.minaserver.module;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ServerChannel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.doeis.common.utils.DateUtils;
import com.doeis.common.utils.StringUtils;
import com.doeis.log.QueueManager;
import com.doeis.minaserver.NettyServer;
import com.doeis.minaserver.SystemProperties;
import com.doeis.minaserver.container.BasicModule;
import com.doeis.minaserver.handler.IdlChannelHandlerAdapter;
import com.doeis.minaserver.handler.PacketHandler;
import com.doeis.minaserver.packet.NettyMessageDecoder;
import com.doeis.minaserver.packet.NettyMessageEncoder;
import com.doeis.minaserver.packet.Packet;
import com.doeis.minaserver.session.ClientConnection;
import com.doeis.uwork.common.Constant;
import com.doeis.uwork.dao.MinaSessionLogMongoDao;
import com.doeis.uwork.entity.mongo.MinaActiveSessionLog;
import com.doeis.uwork.entity.mongo.MinaSessionCountLog;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.voland.core.utils.IDGenerator;


public class SocketAcceptorManager extends BasicModule {
	
	/**
	 * 与客户端链接最长时间
	 */
	private static final int MAX_IDLE_TIME = 5 * 60; // 单位是秒
	//由于超时时间过长，10W个长链接链路会创建10W个ScheduledFutureTask对象，
	//每个对象还保存有业务的成员变量，非常消耗内存。用户的持久代设置的比较大，一些定时任务被老化到持久代中，
	//没有被JVM垃圾回收掉，内存一直在增长，用户误认为存在内存泄露
	
	//客户端打开的所有链接 ，进行记录
	private final ChannelGroup allChannels = new  DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
	private static final Logger log = Logger.getLogger(SocketAcceptorManager.class);
	
	//session 检测线程
	private SessionHearThread  consumerThread = null;
	private boolean destory = false;//是否销毁
	
	public SocketAcceptorManager(){
		super("Socket Acceptor Manager");
	}
	
	@Override
	public void initialize(NettyServer server) {
		super.server = server;
	}

	@Override
	public void start() {
		// 设置监听程序。启动监听	
		consumerThread = new SessionHearThread();
		destory = false;
		consumerThread.start();
		bind();
	}

	@Override
	public void stop() {
		// 停止监听，释放资源
		 if(bossGroup!=null){
			 bossGroup.shutdownGracefully();
		 }
		 if(workerGroup!=null){
			 workerGroup.shutdownGracefully();
		 }
		 destory = true;
	}
	
	private EventLoopGroup bossGroup = null;
	private EventLoopGroup workerGroup = null;
	
	public void bind()  {
		
		new Thread(){
			@Override
			public void run() {
				////获取当前系统的CPU数目*2 
				bossGroup = new NioEventLoopGroup(1);
				//bossGroup线程组实际就是Acceptor线程池，负责处理客户端的TCP连接请求，如果系统只有一个服务端端口需要监听，
				//则建议bossGroup线程组线程数设置为1。
				workerGroup = new NioEventLoopGroup(8);
				//workerGroup是真正负责I/O读写操作的线程组，通过ServerBootstrap的group方法进行设置，用于后续的Channel绑定。
				try {
					ServerBootstrap b = new ServerBootstrap();
					b.group(bossGroup, workerGroup)
							.channelFactory(new NettyServerChannelFactory<ServerChannel>(NioServerSocketChannel.class))
							.option(ChannelOption.SO_BACKLOG, 1024)
							.childOption(ChannelOption.SO_KEEPALIVE, true)
							.childHandler(new ChannelInitializer<SocketChannel>() {
								@Override
								protected void initChannel(SocketChannel socketChannel) throws Exception {
									socketChannel.pipeline().addLast(new NettyMessageDecoder());
									socketChannel.pipeline().addLast(new NettyMessageEncoder());
									socketChannel.pipeline().addLast("IdleStateHandler",
											new IdleStateHandler( MAX_IDLE_TIME,  MAX_IDLE_TIME,  MAX_IDLE_TIME ));
									socketChannel.pipeline().addLast("IdlChannelHandlerAdapter",
											new IdlChannelHandlerAdapter());
									socketChannel.pipeline().addLast(new NettyChannelHandlerAdapter());
								}
							});
					System.err.println(" 绑定端口>>"+SystemProperties.getSystemHost()+":"+ SystemProperties.getSystemPort() );
					// 绑定端口，同步等待成功
					b.bind(SystemProperties.getSystemHost(), SystemProperties.getSystemPort()).sync().channel().closeFuture().sync();
					// 等待服务端监听端口关闭
//					channelFuture.channel().closeFuture()
//					channelFuture.channel().closeFuture().sync();
//					channelFuture.channel().closeFuture().syncUninterruptibly();
				} catch (Exception e) {
					e.printStackTrace();
				}finally {
					// 优雅退出，释放线程池资源
					bossGroup.shutdownGracefully();
					workerGroup.shutdownGracefully();
				}
			}
		}.start();
		
	 }
	
	
	/*** 与客户端的链接管理	 */
	public static final String CONNECTION = "_client_conn";
	/**
	 * 主要的业务处理类
	 * **/
	class NettyChannelHandlerAdapter extends ChannelHandlerAdapter {
		//全局变量
		PacketHandler packetHandler = new PacketHandler(server.getPacketRouter());
		
		private final Logger log = Logger.getLogger(NettyChannelHandlerAdapter.class);
		
		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg)
				throws Exception {
			if(SystemProperties.isLogError()){
				log.error( " ctx.channel().id(): " + ctx.channel().id().toString() + " >>> "+msg.toString());
			}
			
			Packet packet = (Packet)msg;
			String from = packet.getFrom();
			String to = packet.getTo();
			
			packet.setFirstTime(com.voland.core.utils.DateUtils.currentDateTime());
			
			if( StringUtils.isBlank(from) || StringUtils.isBlank(to) ){
				throw new NullPointerException("From or to with null value!");
			}
			if( !from.equals(to) ){	
				//这里把业务数据上传到用户层，用户来处理实际业务
				AttributeKey<ClientConnection> attributeKey = AttributeKey.valueOf(CONNECTION); 
				Attribute<ClientConnection> attribute = ctx.attr(attributeKey);
				ClientConnection clienConn = attribute.get();
				packetHandler.process(packet, clienConn, server);
			}
		}

		
		@Override
		public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
			if(SystemProperties.isLogError()){
				log.error( " ctx.channel().id(): " + ctx.channel().id().toString());
			}
			super.channelRegistered(ctx);
			
			allChannels.add( ctx.channel());
			
			sendLog(ctx, Constant.SESSION_STATUS_OPEN);
			ClientConnectionManager connectionManager = server.getClientConnectionManager();
			
			// 生成ClientConnection
			ClientConnection clienConnection = new ClientConnection(ctx, connectionManager.next());	
			connectionManager.addConnection(clienConnection);
			
			AttributeKey<ClientConnection> connectionKey = AttributeKey.valueOf(CONNECTION); 
			Attribute<ClientConnection> conAttribute = ctx.attr(connectionKey);
			conAttribute.set(clienConnection);
			
		}

		@Override
		public void channelInactive(ChannelHandlerContext ctx) throws Exception {
			if(SystemProperties.isLogError()){
				log.error( " ctx.channel().id(): " + ctx.channel().id().toString());
			}
			super.channelInactive(ctx);
			sendLog(ctx, Constant.SESSION_STATUS_CLOSE);
			closeClientConnection(ctx);
			allChannels.remove(ctx.channel());
		}

		@Override
		public void close(ChannelHandlerContext ctx, ChannelPromise promise)
				throws Exception {
			super.close(ctx, promise);
			closeClientConnection(ctx);
			allChannels.remove(ctx.channel());
		}

		@Override
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
			log.error( " ctx.channel().id(): " + ctx.channel().id().toString());
			
			AttributeKey<ClientConnection> attributeKey = AttributeKey.valueOf(CONNECTION); 
			Attribute<ClientConnection> attribute = ctx.attr(attributeKey);
			ClientConnection clienConn = attribute.get();
			clienConn.close();
			
			String msg = cause.getMessage();		
			if (cause instanceof IOException) {
//				log.error(msg, cause);
			}else {
				log.error(msg, cause);
			}	
			ctx.close();
			allChannels.remove(ctx.channel());
		}
		
		
		/**
		 * 任何连接被关闭的情况都会调用本方法
		 * @param ioSession
		 */
		private void closeClientConnection(ChannelHandlerContext ctx) {
			AttributeKey<ClientConnection> attributeKey = AttributeKey.valueOf(CONNECTION); 
			Attribute<ClientConnection> attribute = ctx.attr(attributeKey);
			ClientConnection clienConn = attribute.get();
			if(clienConn==null){
				return ;
			}
			clienConn.setStatus(ClientConnection.STATUS_CLOSED);
			
			ClientConnectionManager connectionManager = server.getClientConnectionManager();
			connectionManager.removeConnectionAfterClose(clienConn);			
		}
		
		
		private void sendLog(ChannelHandlerContext session, int status){
			QueueManager queueManager = server.getT(QueueManager.class);
			if(queueManager!=null){
				queueManager.sendLog(session, status);
			}
		}
	}
	

	/**
	 * 类介绍     session监听类
	 * @author tianbx
	 * @version 2016年5月21日下午3:31:40
	 */
	class SessionHearThread extends Thread{
		
		MinaSessionLogMongoDao logMongoDao = null;
		QueueManager queueManager = null;
		
		public SessionHearThread() {
			log.error("SessionHearThread 监听创建 ... ");
			Injector injector = Guice.createInjector();
			logMongoDao = injector.getInstance(MinaSessionLogMongoDao.class);
			queueManager = server.getT(QueueManager.class);
		}
		
	    @Override
	    public void run() {
	    	while(!destory){
	    		try {
	    			if(workerGroup==null){
	    				sleep( 1000 * 60 * 3 );//60s * 3
	    			}
	    			
	    			ClientConnectionManager	connectManager = server.getClientConnectionManager();
	    			
	    			String data_group = IDGenerator.uuid();
		    		MinaSessionCountLog sessionCountLog = new MinaSessionCountLog();
		    		sessionCountLog.setId(IDGenerator.uuid()); 
		    		sessionCountLog.setMinacount(allChannels.size());
		    		sessionCountLog.setCreated_at(DateUtils.currentDateTime());
		    		sessionCountLog.setData_group(data_group);
		    		
		    		Map<String, ClientConnection>  authMap = connectManager.getAuthConnections();
		    		sessionCountLog.setAuthcount(authMap.size());
		    		
		    		Map<String, ClientConnection>  connMap =  connectManager.getConnections();
		    		sessionCountLog.setConncount(connMap.size());
		    		
		    		queueManager.sendLog( sessionCountLog);
		    		
		    		MinaActiveSessionLog activeSessionLog = null;
		    		ChannelHandlerContext ioSession = null;
		    		List<ClientConnection> listValue = new ArrayList<ClientConnection>(authMap.values());
		    		
		    		for(ClientConnection connection : listValue ){
		    			activeSessionLog = new MinaActiveSessionLog();
			    		activeSessionLog.setCreated_at(DateUtils.currentDateTime());
			    		activeSessionLog.setData_group(data_group);
		    			activeSessionLog.setId(IDGenerator.uuid());
		    			activeSessionLog.setUser_id(connection.getUserid());
		    			
		    			ioSession = connection.getChannelHandlerContext();
		    			if(ioSession!=null){
		    				activeSessionLog.setSessionid(ioSession.channel().id().asShortText());
		    				SocketAddress socketAddress = ioSession.channel().remoteAddress();
		    				if(socketAddress instanceof InetSocketAddress){
		    					InetSocketAddress inSocketAddress = (InetSocketAddress) socketAddress;
		    					activeSessionLog.setRemoteIp(inSocketAddress.getHostName());
		    					activeSessionLog.setRemotePort(inSocketAddress.getPort());
		    				}
		    			}
		    			activeSessionLog.setAuth(true);
		    			queueManager.sendLog( activeSessionLog);
		    		}
		    		
		    		listValue.clear();
		    		listValue.addAll(connMap.values());
		    		for(ClientConnection connection : listValue ){
		    			
		    			activeSessionLog = new MinaActiveSessionLog();
			    		activeSessionLog.setCreated_at(DateUtils.currentDateTime());
			    		activeSessionLog.setData_group(data_group);
			    		
		    			activeSessionLog.setId(IDGenerator.uuid());
		    			activeSessionLog.setUser_id(connection.getUserid());
		    			
		    			ioSession = connection.getChannelHandlerContext();
		    			if(ioSession!=null){
		    				activeSessionLog.setSessionid(ioSession.channel().id().asShortText());
		    				SocketAddress socketAddress = ioSession.channel().remoteAddress();
		    				if(socketAddress instanceof InetSocketAddress){
		    					InetSocketAddress inSocketAddress = (InetSocketAddress) socketAddress;
		    					activeSessionLog.setRemoteIp(inSocketAddress.getHostName());
		    					activeSessionLog.setRemotePort(inSocketAddress.getPort());
		    				}
		    			}
		    			activeSessionLog.setAuth(false);
		    			queueManager.sendLog( activeSessionLog);
		    		}
				} catch (Exception e) {
					e.printStackTrace();
				}finally{
					try {
						sleep( 1000 * 60 );//60s	
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
	    	}
	    }
	 
	}
}
